package com.atguigu.tran;

import com.atguigu.bean.WaterSensor;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


import java.util.Arrays;
import java.util.List;

/**
 * @author yhm
 * @create 2024-04-02 10:07
 */
public class Test03_FlatMap {
    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",8081);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        // 2. 读取数据源
        List<WaterSensor> waterSensors = Arrays.asList(new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_2", 2000L, 20),
                new WaterSensor("sensor_3", 3000L, 30),
                new WaterSensor("sensor_4", 4000L, 40)
        );
        DataStreamSource<WaterSensor> dataStreamSource = env.fromCollection(waterSensors);

        // 3. 处理数据
        // 如果id为sensor_1 下游输出vc  -> map
        // 如果id为sensor_2 下游输出ts和vc  -> 炸裂
        // 如果id为其他  过滤掉  -> filter
        SingleOutputStreamOperator<String> streamOperator = dataStreamSource.flatMap(new FlatMapFunction<WaterSensor, String>() {
            @Override
            public void flatMap(WaterSensor value, Collector<String> out) throws Exception {
                if ("sensor_1".equals(value.id)) {
                    out.collect(value.vc.toString());
                } else if ("sensor_2".equals(value.id)) {
                    out.collect(value.ts.toString());
                    out.collect(value.vc.toString());
                } else {
                    // 过滤掉 不处理
                }
            }
        });
        // lambda表达式
        SingleOutputStreamOperator<String> streamOperator1 = dataStreamSource.flatMap((WaterSensor value, Collector<String> out) -> {
            if ("sensor_1".equals(value.id)) {
                out.collect(value.vc.toString());
            } else if ("sensor_2".equals(value.id)) {
                out.collect(value.ts.toString());
                out.collect(value.vc.toString());
            } else {
                // 过滤掉 不处理
            }
        }).returns(Types.STRING);

        // 使用子类实现对应的接口
        dataStreamSource.flatMap(new CustomFlatMapFunction()).print();


        // 4. 输出

        // 5. 执行环境
        env.execute();
    }

    public static class CustomFlatMapFunction implements FlatMapFunction<WaterSensor,String>{

        @Override
        public void flatMap(WaterSensor value, Collector<String> out) throws Exception {
            if ("sensor_1".equals(value.id)) {
                out.collect(value.vc.toString());
            } else if ("sensor_2".equals(value.id)) {
                out.collect(value.ts.toString());
                out.collect(value.vc.toString());
            } else {
                // 过滤掉 不处理
            }
        }
    }
}
